package com.atguigu.flink.sql.function;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Expressions;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2023/4/14
 *
 *  表值聚合函数 将多行数据里的标量值转换成新的行数据；
 *      本质上是 UDTF，但是生成的结果是聚合后的结果。
 *
 *
 *  求每种传感器，前两名的vc
 *      输入:  s1,100,10  输出:  s1, 10 , 1
 *      输入:  s1,100,20  输出:  s1, 20 , 1
 *                              s1, 10, 2
 *     输入:  s1,100,30  输出:  s1, 30 , 1
 *                             s1, 20 , 2
 *
 *
 *   ①按照flink的规范，定义一个函数类
 *   ②在系统中注册函数，提供一个函数名字
 *   ③可以调用函数
 */
public class Demo6_CustomUDTAFFunciton
{
    public static void main(String[] args) {

        EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance().inStreamingMode().build();
        TableEnvironment tableEnvironment = TableEnvironment.create(environmentSettings);

        String createTableSql = " create table t1 ( id STRING, ts BIGINT , vc INT " +
            "    )WITH (" +
            "  'connector' = 'filesystem'," +
            "  'path' = 'data/sensor.txt'," +
            "  'format' = 'csv'" +
            ") ";

        tableEnvironment.executeSql(createTableSql);

        //②在系统中注册函数，提供一个函数名字
        MyTop2Vc myFunction = new MyTop2Vc();
        tableEnvironment.createTemporaryFunction("myTop2Vc",myFunction);

        /*
            在标准的sql中，就没有  UDTAF这种语法！
                flink中编写的sql，使用 apache calilte 解析，只支持标准sql。
                sql中无法使用，可以使用tableapi操作。
         */

        //基于表名 获取 Table对象
        Table t1 = tableEnvironment.from("t1");

        t1
            .groupBy($("id"))
            .flatAggregate(Expressions.call("myTop2Vc",$("vc")))
            .select($("id"),$("rank"),$("vcNum"))
            .execute()
            .print();


    }

    /*
        你需要扩展  TableAggregateFunction<T, ACC> 类

           下面几个 TableAggregateFunction 的方法是必须要实现的：

                createAccumulator()
                accumulate()
     */

    @FunctionHint(output = @DataTypeHint("ROW<rank INT, vcNum INT>"))
    public static class MyTop2Vc extends TableAggregateFunction<Row,Top2Vc>
    {
        /*
            输出结果
                参数1： 累加器
                参数2： Collector<T>
         */
        public void emitValue(Top2Vc acc, Collector<Row> out){
            out.collect(Row.of(1,acc.firstVc));
            //out.collect()调用一次，输出一行
            if (acc.getSecondVc() > 0){
                //说明有第二名
                out.collect(Row.of(2,acc.secondVc));
            }

        }
        /*
                参数1，是累加器
                参数2，函数中传入的列的类型
         */
        public void accumulate(Top2Vc acc, Integer vc) {

            //当前水位比 之前累积的第一名小
            if (vc <= acc.firstVc){
                //和第二名比较
                if (vc > acc.secondVc){
                    acc.setSecondVc(vc);
                }
            }else {

                //把之前的第一名降级为第二名，把当前水位设置为第一名
                acc.setSecondVc(acc.getFirstVc());
                acc.setFirstVc(vc);

            }
        }

        @Override
        public Top2Vc createAccumulator() {
            return new Top2Vc(0,0);
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Top2Vc{

        //存储某种传感器的前两名的vc
        private Integer firstVc;
        private Integer secondVc;
     }
}
